﻿#include <osa_mqueue.h>

int OSA_mqueCreate(OSA_MqueHndl *hndl, char* name, Uint32 maxLen, Uint32 maxSize, Uint32 block)
{
	pthread_mutexattr_t mutex_attr;
	pthread_condattr_t cond_attr;
	int status = OSA_SOK;
	 
	hndl->curRd = hndl->curWr = 0;
	hndl->count = 0;
	hndl->len = maxLen;
	hndl->size = maxSize;
	hndl->block = block;
	memset(hndl->name, 0, sizeof(hndl->name));
	memcpy(hndl->name, name, strlen(name));
	hndl->queue = (char*)OSA_memAlloc(hndl->len*hndl->size);

	if (hndl->queue == NULL) {
		OSA_ERROR("OSA_queCreate() = %d \r\n", status);
		return OSA_EFAIL;
	}

	status |= pthread_mutexattr_init(&mutex_attr);
	status |= pthread_condattr_init(&cond_attr);

	status |= pthread_mutex_init(&hndl->lock, &mutex_attr);
	status |= pthread_cond_init(&hndl->condRd, &cond_attr);
	status |= pthread_cond_init(&hndl->condWr, &cond_attr);

	if (status != OSA_SOK)
		OSA_ERROR("OSA_queCreate() = %d \r\n", status);

	pthread_condattr_destroy(&cond_attr);
	pthread_mutexattr_destroy(&mutex_attr);

	return status;
}

int OSA_mqueDelete(OSA_MqueHndl *hndl)
{
	if (hndl->queue != NULL)
		OSA_memFree(hndl->queue);

	pthread_cond_destroy(&hndl->condRd);
	pthread_cond_destroy(&hndl->condWr);
	pthread_mutex_destroy(&hndl->lock);

	return OSA_SOK;
}

int OSA_mquePut(OSA_MqueHndl *hndl, char *buf, Uint32 size)
{
	int status = OSA_EFAIL;

	if (size > hndl->size)
		return status;

	pthread_mutex_lock(&hndl->lock);

	while (1) {
		if (hndl->count < hndl->len) {
			memcpy(hndl->queue + hndl->curWr*hndl->size, buf, size);
			memset(hndl->queue + hndl->curWr*hndl->size + size, 0, hndl->size - size);
			hndl->curWr = (hndl->curWr + 1) % hndl->len;
			hndl->count++;
			status = OSA_SOK;
			pthread_cond_signal(&hndl->condRd);
			break;
		}
		else {
			if (!hndl->block)
				break;

			status = pthread_cond_wait(&hndl->condWr, &hndl->lock);
		}
	}

	pthread_mutex_unlock(&hndl->lock);

	return status;
}

int OSA_mqueGet(OSA_MqueHndl *hndl, char *buf, Uint32 size)
{
	int status = OSA_EFAIL;

	if (size > hndl->size)
		return status;

	pthread_mutex_lock(&hndl->lock);

	while (1) {
		if (hndl->count > 0) {

			if (buf != NULL) {
				memcpy(buf, hndl->queue + hndl->curRd * hndl->size, size);
			}

			hndl->curRd = (hndl->curRd + 1) % hndl->len;
			hndl->count--;
			status = OSA_SOK;
			pthread_cond_signal(&hndl->condWr);
			break;
		} 
		else {
			if (!hndl->block)
				break;

			status = pthread_cond_wait(&hndl->condRd, &hndl->lock);
		}
	}

	pthread_mutex_unlock(&hndl->lock);

	return status;
}

Uint32 OSA_mqueGetQueuedCount(OSA_MqueHndl *hndl)
{
	Uint32 queuedCount = 0;

	pthread_mutex_lock(&hndl->lock);
	queuedCount = hndl->count;
	pthread_mutex_unlock(&hndl->lock);
	return queuedCount;
}

int OSA_mqueIsEmpty(OSA_MqueHndl *hndl)
{
	int isEmpty;

	pthread_mutex_lock(&hndl->lock);
	if (hndl->count == 0)
	{
		isEmpty = TRUE;
	}
	else
	{
		isEmpty = FALSE;
	}
	pthread_mutex_unlock(&hndl->lock);

	return isEmpty;
}
